package com.jiajun.ct.analysis.io;

import com.jiajun.common.util.JDBCUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

/**
 * MySQL数据的格式化输入对象
 */
public class MySQLTextOutputFormat extends OutputFormat<Text,Text> {

    protected static class MySQLRecordWriter extends RecordWriter<org.apache.hadoop.io.Text, org.apache.hadoop.io.Text>{

        private Connection connection = null;

        private Jedis jedis = null;

        //构造方法
        public MySQLRecordWriter(){
            //获取资源
            connection = JDBCUtil.getConnection();
            jedis = new Jedis("192.168.30.102",6379);
            jedis.auth("000000");
        }

        //输出数据
        public void write(Text text, Text text2) throws IOException, InterruptedException {
            String insertSQL = "insert into ct_call (telid,dateid,sumcall,sumduration) " +
                    "values (?,?,?,?)";

            PreparedStatement pstat = null;

            String[] values = text2.toString().split("_");
            String sumcall = values[0];
            String sumduration = values[1];

            try {
                pstat = connection.prepareStatement(insertSQL);

                String k = text.toString();
                String[] ks = k.split("_");

                String tel = ks[0];
                String date = ks[1];

                pstat.setInt(1,Integer.parseInt(jedis.hget("ct_user",tel)));
                pstat.setInt(2,Integer.parseInt(jedis.hget("ct_date",date)));
                pstat.setInt(3,Integer.parseInt(sumcall));
                pstat.setInt(4,Integer.parseInt(sumduration));
                pstat.executeUpdate();

            } catch (SQLException e) {
                e.printStackTrace();
            }finally {
                if(pstat!=null){
                    try {
                        pstat.close();
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        //释放资源
        public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
            if( connection != null ){
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
            if( jedis!=null ){
                jedis.close();
            }
        }
    }

    //OutputFormat就是靠getReccordWriter对象将数据输出到MySQL中
    public RecordWriter<Text,Text> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        return new MySQLRecordWriter();
    }

    public void checkOutputSpecs(JobContext jobContext) throws IOException, InterruptedException {

    }

    //源码模仿
    private FileOutputCommitter committer = null;
    public static Path getOutputPath(JobContext job){
        String name = job.getConfiguration().get(FileOutputFormat.OUTDIR);
        return name == null ? null : new Path(name);

    }

    //提交MR任务给hadoop,逻辑不需要的时候打开源码模仿
    public OutputCommitter getOutputCommitter(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        if(committer ==null){
            Path output = getOutputPath(taskAttemptContext);
            committer = new FileOutputCommitter(output,taskAttemptContext);
        }
        return committer;
    }
}
